#include "ShmQueueTestSuite.h"
#include <cstdint>
#include <gtest/gtest.h>
#include <thread>
#include "FIFOQueue.h"
#include "XnccType.h"
#include "xncc_random.hpp"

TestQueue*                   SHMQueueTestSuite::shm_queue_producer = nullptr;
TestQueue*                   SHMQueueTestSuite::shm_queue_consumer = nullptr;
xncc::shared::MmapSHMObject* SHMQueueTestSuite::shm_producer       = nullptr;
xncc::shared::MmapSHMObject* SHMQueueTestSuite::shm_consumer       = nullptr;

static constexpr uint32_t THREAD_MSG_HEAD_LEN = sizeof(xncc::msg::MsgHeaderInThread);

TEST_F(SHMQueueTestSuite, thread_producer_consumer)
{
    xncc::tools::xncc_random<xncc::types::queue_size_t> random(24, 1000);
    xncc::tools::xncc_random<std::string>               randomStr;

    auto func = [&random, &randomStr](TestQueue* queue) {
        for (int i = 0; i < 1000000; i++) {
            const auto len = random.get_value();
            const auto buf = randomStr.get_value(int(len));
            while (true) {
                char* writeBuf = queue->getWriteBuffer(THREAD_MSG_HEAD_LEN + len);
                if (writeBuf) {
                    xncc::msg::MsgHeaderInThread* hptr = reinterpret_cast<xncc::msg::MsgHeaderInThread*>(writeBuf);
                    hptr->len                          = static_cast<uint32_t>(len);
                    hptr->id                           = 1000;
                    hptr->connId                       = 1;
                    hptr->tid                          = 1;
                    memcpy(writeBuf + THREAD_MSG_HEAD_LEN, buf.c_str(), static_cast<uint32_t>(len));
                    queue->finishWrite();
                    break;
                }
            }
        }
    };

    std::thread t1(func, shm_queue_producer);

    int i = 0;
    while (true) {
        char* readBuf = shm_queue_consumer->getReadBuffer(THREAD_MSG_HEAD_LEN);
        if (unlikely(readBuf == nullptr)) {
            continue;
        }
        xncc::msg::MsgHeaderInThread* head = reinterpret_cast<xncc::msg::MsgHeaderInThread*>(readBuf);
        EXPECT_EQ(head->id, 1000);
        EXPECT_EQ(head->connId, 1);
        EXPECT_EQ(head->tid, 1);

        auto total_len = head->len + THREAD_MSG_HEAD_LEN;
        ++i;

        char* dataBuf =
            shm_queue_consumer->getReadBuffer(static_cast<xncc::types::queue_size_t>(head->len) + THREAD_MSG_HEAD_LEN);
        EXPECT_TRUE(dataBuf != nullptr);
        if (likely(dataBuf)) {
            shm_queue_consumer->finishRead();
        }
        else {
            std::cout << "read second nullptr i = " << i << " len " << total_len << std::endl;
        }
        if (i % 100000 == 0) {
            std::cout << "index:" << i << " data_len:" << total_len << std::endl;
        }
        if (i >= 1000000) {
            break;
        }
    }

    EXPECT_TRUE(i == 1000000);
    t1.join();
}